package com.lz.kafka.consumer.project.kafka.service;

import cn.hutool.core.date.DateUtil;
import com.lz.kafka.consumer.common.ResponseVo;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
/**
 * Kafka 消费
 *
 * */
@Component
public class KafkaConsumerService {
    public static Object vv ;

  /*  @KafkaListener(groupId = "group01",topics = "topicfirst")
    public void onMessage(ConsumerRecord<String,Object> record, Acknowledgment acknowledgment, Consumer<?,?> consumer){
        System.out.println("*****消费者 获取的消息: "+record.value());
        vv = record.value();
        acknowledgment.acknowledge();
    }
*/
    @KafkaListener(groupId = "group01",topics = "topicfirst")
    public void onMessage2(List<ConsumerRecord<String,Object>> records, Acknowledgment acknowledgment, Consumer<?,?> consumer){
        for (int i = 0; i < records.size(); i++) {
            System.out.println(DateUtil.date() +"*****消费者 获取的消息: "+records.get(i).value());
        }
        vv = records.get(0).value();
        acknowledgment.acknowledge();
    }

    public ResponseVo getReceive(){
        return ResponseVo.success(vv);
    }
}
